// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/uniswap/feed_v3/processor.h"

#include <limits>
#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/uniswap/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::uniswap::feed_v3::impl {

void UniswapFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  const auto& data = packet.payload;

  rapidjson::Document document;

  document.Parse(data.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  int worker_id = 0;
  if (document.HasMember("topic") && document.HasMember("worker_id")) {
    worker_id = document["worker_id"].GetInt();
    document.Parse(document["data"].GetString());
    if (document.HasParseError()) {
      THROW() << "JSON parse error: "
              << GetParseError_En(document.GetParseError()) << packet.payload;
    }
  }
  std::string_view native_symbol = document["symbol"].GetString();
  if (document.HasMember("bid0") &&
      document.HasMember("ask0")) {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      UniswapFeedUpdateParser::ParseBbo(book, document, packet.timestamp);
      upd_mgr_.PublishBook();
    }
  } else if (document.HasMember("trades")) {
    // Only handle either worker1 or worker0;
    if (worker_id == 2) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
    auto trades_size = document["trades"].Size();
    int i = 0;

    const auto* book = upd_mgr_.GetBook(native_symbol);
    double valid_price_floor = 0;
    double valid_price_ceil = std::numeric_limits<double>::max();
    if (book != nullptr) {
      valid_price_floor = book->Bid0()->price * (1 - 0.2);
      valid_price_ceil = book->Ask0()->price * (1 + 0.2);
    }
    for (const auto& trade_json : document["trades"].GetArray()) {
      Trade* trade = upd_mgr_.GetTradeMutable();
      UniswapFeedUpdateParser::ParseTrade(trade_json, trade, packet.timestamp);
      if (trade->price <= valid_price_floor ||
          trade->price >= valid_price_ceil) {
        ++i;
        LOG(ERROR) << "[Uniswap Feed Processor] filter out invalid price "
                   << native_symbol << ": " << trade->price
                   << "\n" << packet.payload.data();
        continue;
      }
      if (i < trades_size - 1) {
        upd_mgr_.PublishTrade(native_symbol, true);
      } else {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
      ++i;
    }
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::uniswap::feed_v3::impl
